package com.hsyt.pfc.redis.serializer;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.SerializationException;

import java.io.ByteArrayOutputStream;
import java.util.Collections;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @Description: kryo序列化
 * @ClassName: KryoRedisSerializer
 * @Package: com.hsyt.pfc.redis.serializer
 * @Author: wangze@hsyuntai.com
 * @Date: 2018年9月18日 16:09
 * @Copyright: 版权归 HSYUNTAI 所有
 */
public class KryoRedisSerializer<T> implements RedisSerializer<T> {

    public static final byte[] EMPTY_BYTE_ARRAY = new byte[0];

    public interface KryoPool {

        Kryo get();

        void yield(Kryo kryo);

    }

    public static class KryoPoolImpl implements KryoPool {

        private final Queue<Kryo> objects = new ConcurrentLinkedQueue<Kryo>();
        private final List<Class<?>> classes;

        public KryoPoolImpl(List<Class<?>> classes) {
            this.classes = classes;
        }

        public Kryo get() {
            Kryo kryo;
            if ((kryo = objects.poll()) == null) {
                kryo = createInstance();
            }
            return kryo;
        }

        public void yield(Kryo kryo) {
            objects.offer(kryo);
        }

        protected Kryo createInstance() {
            Kryo kryo = new Kryo();
            kryo.setReferences(false);
            for (Class<?> clazz : classes) {
                kryo.register(clazz);
            }
            return kryo;
        }

    }

    private final KryoPool kryoPool;

    @SuppressWarnings("unchecked")
    public T deserialize(byte[] buf) throws SerializationException {
        Kryo kryo = null;
        try {
            kryo = kryoPool.get();
            return (T) kryo.readClassAndObject(new Input(buf));
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        } finally {
            if (kryo != null) {
                kryoPool.yield(kryo);
            }
        }
    }

    public byte[] serialize(Object in) throws SerializationException {
        Kryo kryo = null;
        try {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            Output output = new Output(baos);
            kryo = kryoPool.get();
            kryo.writeClassAndObject(output, in);
            output.close();
            return baos.toByteArray();
        } catch (Exception e) {
            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            }
            e.printStackTrace();
        } finally {
            if (kryo != null) {
                kryoPool.yield(kryo);
            }
        }
        return null;
    }

    public KryoRedisSerializer() {
        this(new KryoPoolImpl(Collections.<Class<?>>emptyList()));
    }

    public KryoRedisSerializer(List<Class<?>> classes) {
        this(new KryoPoolImpl(classes));
    }

    public KryoRedisSerializer(KryoPool kryoPool) {
        this.kryoPool = kryoPool;
    }

}
